package FutureAndPromise.Demo05_Comparison;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.*;

/**
 * Demo05 - Future 和 Promise 全面对比
 * 
 * 学习目标：
 * 1. 对比 JDK Future、CompletableFuture 和 Netty Future
 * 2. 理解 Promise 的生产者-消费者模式
 * 3. 掌握不同场景下的最佳选择
 */
@Slf4j
public class Client {

    public static void main(String[] args) throws Exception {
        log.info("=== Future 和 Promise 全面对比 ===\n");
        
        demo1_JDKFutureVsNettyFuture();
        demo2_CompletableFutureVsNettyFuture();
        demo3_FutureVsPromise();
        demo4_RealWorldScenarios();
        
        Thread.sleep(2000);
        System.exit(0);
    }

    /**
     * Demo 1: JDK Future vs Netty Future
     */
    private static void demo1_JDKFutureVsNettyFuture() throws Exception {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 1: JDK Future vs Netty Future");
        log.info("└─────────────────────────────────────────\n");
        
        // 1. JDK Future - 只能阻塞等待
        log.info("1. JDK Future 方式（阻塞）：");
        ExecutorService executor = Executors.newCachedThreadPool();
        
        Future<String> jdkFuture = executor.submit(() -> {
            log.info("   [JDK] 执行任务...");
            Thread.sleep(500);
            return "JDK Future 结果";
        });
        
        log.info("   [JDK] 提交任务后立即返回");
        log.info("   [JDK] 阻塞等待结果...");
        String result = jdkFuture.get(); // 阻塞！
        log.info("   [JDK] 得到结果: {}", result);
        
        executor.shutdown();
        
        // 2. Netty Future - 可以使用监听器
        log.info("\n2. Netty Future 方式（非阻塞）：");
        EventLoopGroup group = new NioEventLoopGroup();
        
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new SimpleHandler());
                    }
                });
            
            log.info("   [Netty] 发起连接...");
            ChannelFuture nettyFuture = b.connect("localhost", 9000);
            
            log.info("   [Netty] 立即返回，添加监听器");
            nettyFuture.addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    log.info("   [Netty] ✅ 监听器回调 - 连接成功！");
                    future.channel().close();
                }
            });
            
            log.info("   [Netty] 主线程继续执行，不阻塞");
            
            Thread.sleep(1000);
            
        } finally {
            group.shutdownGracefully();
        }
    }

    /**
     * Demo 2: CompletableFuture vs Netty Future
     */
    private static void demo2_CompletableFutureVsNettyFuture() throws Exception {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 2: CompletableFuture vs Netty Future");
        log.info("└─────────────────────────────────────────\n");
        
        // 1. CompletableFuture - JDK 8 的异步方案
        log.info("1. CompletableFuture 方式：");
        
        CompletableFuture.supplyAsync(() -> {
            log.info("   [CF] 步骤1: 获取数据");
            return "数据";
        }).thenApply(data -> {
            log.info("   [CF] 步骤2: 处理数据 '{}'", data);
            return data.toUpperCase();
        }).thenAccept(result -> {
            log.info("   [CF] 步骤3: 结果 '{}'", result);
        }).exceptionally(ex -> {
            log.error("   [CF] 异常处理");
            return null;
        });
        
        Thread.sleep(500);
        
        // 2. Netty Future - Netty I/O 操作
        log.info("\n2. Netty Future 方式：");
        EventLoopGroup group = new NioEventLoopGroup();
        
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new SimpleHandler());
                    }
                });
            
            b.connect("localhost", 9000)
                .addListener((ChannelFutureListener) connectFuture -> {
                    if (connectFuture.isSuccess()) {
                        log.info("   [Netty] 步骤1: 连接成功");
                        Channel channel = connectFuture.channel();
                        
                        channel.writeAndFlush(
                            Unpooled.copiedBuffer("Test\n", StandardCharsets.UTF_8)
                        ).addListener((ChannelFutureListener) writeFuture -> {
                            if (writeFuture.isSuccess()) {
                                log.info("   [Netty] 步骤2: 发送成功");
                                channel.close().addListener(closeFuture -> {
                                    log.info("   [Netty] 步骤3: 关闭完成");
                                });
                            }
                        });
                    }
                });
            
            Thread.sleep(1000);
            
        } finally {
            group.shutdownGracefully();
        }
    }

    /**
     * Demo 3: Future vs Promise 对比
     */
    private static void demo3_FutureVsPromise() throws Exception {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 3: Future (只读) vs Promise (可写)");
        log.info("└─────────────────────────────────────────\n");
        
        EventLoopGroup group = new NioEventLoopGroup();
        
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new SimpleHandler());
                    }
                });
            
            ChannelFuture connectFuture = b.connect("localhost", 9000).sync();
            Channel channel = connectFuture.channel();
            
            // 创建一个 Promise
            ChannelPromise promise = channel.newPromise();
            
            log.info("1. Promise（生产者视角）：");
            log.info("   可以主动设置结果");
            
            // 作为 Future 使用（消费者）
            promise.addListener((ChannelFutureListener) future -> {
                log.info("\n3. Future（消费者视角）：");
                log.info("   监听器收到通知");
                if (future.isSuccess()) {
                    log.info("   ✅ 操作成功");
                } else {
                    log.error("   ❌ 操作失败");
                }
            });
            
            // 模拟异步操作
            channel.eventLoop().schedule(() -> {
                log.info("\n2. Promise（生产者视角）：");
                log.info("   设置操作结果为成功");
                promise.setSuccess(); // 设置成功
                // 或者: promise.setFailure(new Exception("失败"));
            }, 500, TimeUnit.MILLISECONDS);
            
            Thread.sleep(1000);
            channel.close().sync();
            
        } finally {
            group.shutdownGracefully();
        }
    }

    /**
     * Demo 4: 实际应用场景对比
     */
    private static void demo4_RealWorldScenarios() throws Exception {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 4: 实际应用场景");
        log.info("└─────────────────────────────────────────\n");
        
        // 场景 1: CPU 密集型任务 - 使用 CompletableFuture
        log.info("场景 1: CPU 密集型计算（推荐 CompletableFuture）");
        CompletableFuture.supplyAsync(() -> {
            log.info("   计算复杂运算...");
            return fibonacci(35);
        }).thenAccept(result -> {
            log.info("   ✅ 计算结果: {}", result);
        });
        
        Thread.sleep(500);
        
        // 场景 2: I/O 操作 - 使用 Netty Future
        log.info("\n场景 2: 网络 I/O 操作（推荐 Netty Future）");
        EventLoopGroup group = new NioEventLoopGroup();
        
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new SimpleHandler());
                    }
                });
            
            b.connect("localhost", 9000).addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    log.info("   ✅ 网络连接建立");
                    future.channel().writeAndFlush(
                        Unpooled.copiedBuffer("Network I/O\n", StandardCharsets.UTF_8)
                    ).addListener(ChannelFutureListener.CLOSE);
                }
            });
            
            Thread.sleep(500);
            
        } finally {
            group.shutdownGracefully();
        }
        
        // 场景 3: 混合场景 - 可以组合使用
        log.info("\n场景 3: 混合场景（可以组合使用）");
        log.info("   示例：先用 CompletableFuture 处理业务逻辑");
        log.info("         再用 Netty Future 发送结果");
        
        CompletableFuture.supplyAsync(() -> {
            // 业务逻辑
            return "处理结果";
        }).thenAccept(result -> {
            log.info("   业务处理完成: {}", result);
            log.info("   可以在这里使用 Netty 发送结果");
        });
        
        Thread.sleep(500);
    }

    // 斐波那契计算（CPU 密集型）
    private static int fibonacci(int n) {
        if (n <= 1) return n;
        return fibonacci(n - 1) + fibonacci(n - 2);
    }

    // 简单的处理器
    @Slf4j
    static class SimpleHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            ByteBuf buf = (ByteBuf) msg;
            log.info("   [客户端] 收到回复: {}", buf.toString(StandardCharsets.UTF_8).trim());
            buf.release();
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            ctx.close();
        }
    }
}

/*
=== 对比总结 ===

┌──────────────────┬──────────────────┬──────────────────┬──────────────────┐
│     特性         │   JDK Future     │CompletableFuture │   Netty Future   │
├──────────────────┼──────────────────┼──────────────────┼──────────────────┤
│ 回调支持         │       ❌         │       ✅         │       ✅         │
│ 监听器           │       ❌         │       ❌         │       ✅         │
│ 链式调用         │       ❌         │       ✅         │       ✅         │
│ 异常处理         │      基础        │      强大        │      中等        │
│ 是否可写         │       ❌         │       ✅         │ ❌(Promise可写)  │
│ 线程模型         │    自定义        │  ForkJoinPool    │   EventLoop      │
│ 使用场景         │    通用异步      │   通用异步       │   Netty I/O      │
└──────────────────┴──────────────────┴──────────────────┴──────────────────┘

=== Future vs Promise ===

Future（只读视角 - 消费者）：
- 查询状态
- 添加监听器
- 等待结果

Promise（可写视角 - 生产者）：
- 设置成功结果
- 设置失败原因
- 通知所有监听器

关系：Promise 继承自 Future，既可以读也可以写

=== 选择建议 ===

1. 通用异步任务（Java 8+）：
   ✅ 使用 CompletableFuture
   - CPU 密集型计算
   - 业务逻辑处理
   - 需要组合多个异步操作

2. Netty 网络 I/O：
   ✅ 使用 Netty Future/Promise
   - 网络连接
   - 数据读写
   - Channel 操作

3. 简单异步任务（Java 5+）：
   ⚠️  可以使用 JDK Future
   - 功能有限
   - 不推荐新项目使用

=== 关键知识点 ===

1. Future 是异步操作结果的抽象
2. Promise 是可写的 Future
3. CompletableFuture 是 JDK 8 的现代异步方案
4. Netty Future 专为 I/O 操作设计
5. 选择合适的工具取决于具体场景

=== 最佳实践 ===

✅ 根据场景选择合适的 Future
✅ 使用监听器代替阻塞等待
✅ 完善的异常处理
✅ 理解 Future 和 Promise 的角色
✅ 避免回调地狱，保持代码清晰
*/

